package ltq.stormOld;

import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.Utils;
import sgt.test.util.DatabaseUtil;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Map;

public class TelDataSpout extends BaseRichSpout {
    private SpoutOutputCollector _collector;
    private Connection con=null;
    private PreparedStatement ps=null;
    private ResultSet rs=null;
    private static int maxid=0;

    @Override
    public void open(Map<String, Object> map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
        _collector=spoutOutputCollector;
        con= DatabaseUtil.getConnection();
        ps=null;
    }

    @Override
    public void nextTuple() {
        try {
            ps=con.prepareStatement("select * from telTest2 where telId>? limit 1000");
            ps.setObject(1,maxid);
            rs=ps.executeQuery();
            while (rs.next()){
                _collector.emit(new Values(rs.getString("telnum1")+" "+rs.getString("telTime")+" 10000"));
                maxid=rs.getInt("telId");
            }
            rs.last();
            int row=rs.getRow();
            if (row==0){
                _collector.emit(new Values("null null 1"));
                Utils.sleep(1000);
            }
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("num_time"));
    }

    @Override
    public void close() {
//        if (rs.row)
    }
}
